package cn.minhx.dtm.starter.core.service.impl;

import cn.minhx.dtm.starter.core.model.InvokeContext;
import cn.minhx.dtm.starter.core.model.TaskStatus;
import cn.minhx.dtm.starter.core.model.TransactionRecord;
import cn.minhx.dtm.starter.core.repository.TransactionRepository;
import cn.minhx.dtm.starter.core.service.TransactionService;
import cn.minhx.dtm.starter.core.transaction.TransactionCommitHandler;
import cn.minhx.dtm.starter.core.util.ReflectionUtils;
import com.alibaba.fastjson2.JSON;
import java.lang.reflect.Method;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.domain.Limit;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.StringUtils;

/**
 * 分布式事务服务实现类
 *
 * @author min
 */
@Service
public class TransactionServiceImpl implements TransactionService {

	private static final Logger log = LoggerFactory.getLogger(TransactionServiceImpl.class);

	private final ThreadPoolExecutor transactionExecutor;
	private final TransactionRepository transactionRepository;
	private final ApplicationContext applicationContext;

	@Autowired
	public TransactionServiceImpl(ThreadPoolExecutor transactionExecutor, TransactionRepository transactionRepository,
			ApplicationContext applicationContext) {
		this.transactionExecutor = transactionExecutor;
		this.transactionRepository = transactionRepository;
		this.applicationContext = applicationContext;
	}

	@Override
	public void executeTransaction(TransactionRecord transaction, boolean async) {
		if (!TransactionSynchronizationManager.isActualTransactionActive()) {
			throw new RuntimeException("必须在事务上下文中执行");
		}
		// 保存事务记录
		transactionRepository.save(transaction);
		// 注册事务完成处理器
		TransactionSynchronizationManager
				.registerSynchronization(new TransactionCommitHandler(() -> execute(transaction, async)));
	}

	@Override
	@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
	public void processTransaction(TransactionRecord transaction) {
		String invokeContext = transaction.getInvokeContext();
		if (!StringUtils.hasText(invokeContext)) {
			String errMsg = "事务记录缺少调用上下文信息";
			log.error(errMsg);
			cancel(transaction, errMsg);
			return;
		}
		InvokeContext methodInvocation = JSON.parseObject(invokeContext, InvokeContext.class);
		String className = methodInvocation.getClassName();
		String methodName = methodInvocation.getMethodName();
		List<Class<?>> paramTypes = ReflectionUtils
				.resolveParameterTypes(List.of(methodInvocation.getParameterTypes()));
		Object[] args = methodInvocation.getArgs();
		try {
			// 更新状态为处理中
			transaction.setStatus(TaskStatus.PROCESSING.getCode());
			transaction = transactionRepository.save(transaction);
			log.debug("状态更新为执行中: {}", JSON.toJSONString(transaction));
			// 获取目标类和方法
			Class<?> clazz = Class.forName(className);
			Method method = ReflectionUtils.findMethod(clazz, methodName, paramTypes.toArray(new Class[0]));
			// 获取目标对象（处理代理对象的情况）
			Object proxy = applicationContext.getBean(clazz);
			Object source = ReflectionUtils.getSourceObject(proxy);
			// 执行方法调用
			method.invoke(source, args);
			// 更新事务状态为成功
			success(transaction);
		} catch (ClassNotFoundException e) {
			log.error("无法找到目标类: {}", className, e);
			// 无法重试的错误，取消
			cancel(transaction, e.getMessage());
		} catch (Exception e) {
			log.error("方法调用失败", e);
			// 可以重试的错误，进行重试
			String failReason = e.getMessage();
			if (!StringUtils.hasLength(failReason)) {
				failReason = e.getCause() == null ? "未知错误" : e.getCause().getMessage();
			}
			retry(transaction, failReason);
		}
	}

	private void cancel(TransactionRecord transaction, String failReason) {
		transaction.setStatus(TaskStatus.CANCELED.getCode());
		transaction.setFailReason(failReason);
		transaction.setRetryCount(transaction.getRetryCount() + 1);
		transactionRepository.save(transaction);
	}

	private void success(TransactionRecord transaction) {
		transaction.setStatus(TaskStatus.SUCCESS.getCode());
		transaction.setRetryCount(transaction.getRetryCount() + 1);
		transactionRepository.save(transaction);
	}

	private void retry(TransactionRecord transaction, String failReason) {
		int currentRetryCount = transaction.getRetryCount() + 1;
		int maxRetryCount = transaction.getMaxRetryCount();

		transaction.setRetryCount(currentRetryCount);
		transaction.setFailReason(failReason);

		if (currentRetryCount >= maxRetryCount) {
			transaction.setStatus(TaskStatus.CANCELED.getCode());
		} else {
			transaction.setNextRetryTime(LocalDateTime.now().plusSeconds(transaction.getRetryInterval()));
			transaction.setStatus(TaskStatus.FAILED.getCode());
		}

		// 更新重试信息
		transactionRepository.save(transaction);
	}

	@Override
	public void retryFailedTransactions(int batchSize) {
		List<TransactionRecord> pendingTransactions = transactionRepository
				.findByStatusInAndNextRetryTimeLessThanEqualAndDeletedFalseOrderByPriorityAscNextRetryTimeAsc(
						List.of(TaskStatus.PENDING.getCode(), TaskStatus.FAILED.getCode()), LocalDateTime.now(),
						Limit.of(batchSize));
		log.info("正在重试失败事务，数量:{} pendingTransactions:{}", pendingTransactions.size(),
				JSON.toJSONString(pendingTransactions));
		pendingTransactions.forEach(r -> execute(r, true));
	}

	private void execute(TransactionRecord transaction, boolean async) {
		TransactionServiceImpl proxy = currentProxy();
		if (async) {
			transactionExecutor.execute(() -> proxy.processTransaction(transaction));
		} else {
			proxy.processTransaction(transaction);
		}
	}

	private TransactionServiceImpl currentProxy() {
		return applicationContext.getBean(this.getClass());
	}
}
